<?php
declare (strict_types=1);

namespace app\command;

use app\common\lib\Redis;
use app\common\lib\ZhyxBeanstalkd;
use app\impl\SimpleFactory;
use Pheanstalk\Exception\ClientException;
use Pheanstalk\Exception\ConnectionException;
use Pheanstalk\Exception\ServerException;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use think\Exception;
use think\facade\Log;

class WatchBean extends Command
{
    protected function configure(): void
    {
        // 指令配置
        $this->setName('watchBean')
            ->setDescription('消费者队列');
    }

    protected function execute(Input $input, Output $output)
    {
        ini_set("default_socket_timeout", "604800");
        ini_set("memory_limit", "1G");
        // $process_num = 4;
        //for ($i = 0; $i < $process_num; $i++) {
        // $pid = pcntl_fork();
        // if ($pid === 0) { 
        $i = 0;
        while (true) {
            $beansTalk = ZhyxBeanstalkd::getInstance()->beansTalk;
            $job = null;
            try {
                $job = $beansTalk
                    ->watch('employeeGranule')
                    ->watch("syncPatientInfo")
                    ->watch("syncOutpatientClinic")
                    ->watch("syncInHospital")
                    ->watch("syncInHospitalOps")
                    ->watch("syncOutHospital")
                    ->watch("syncDiagnosis")
                    ->watch("sendSms")
                    ->watch("sendPropagandaEducation")
                    ->watch("sendFollow")
                    ->watch("sendFollowTask")
                    ->watch("sendDiseasePathPlan")
                    ->watch("tagDiseaseTask")
                    ->watch("pushDiseaseTask")
                    ->watch("sendVisitMessage")
                    ->watch("pushProjectExpMessage")
                    ->watch("sendWxPush")
                    ->ignore("default")->reserve();
                if ($job !== false) {
                    $jobId = (string)$job->getId();
                    $cache_key = 'queue_job:' . $jobId;
                    if (!Redis::getInstance()->redis->keys($cache_key)) {
                        $output->writeln("收到任务进程号-0{$i}:" . $job->getData());
                        $data = json_decode($job->getData(), true);
                        $output->writeln("任务名称" . $data["tubeName"] . "接收任务进程号:0" . $i);
                        $task = SimpleFactory::getTask(ucfirst($data["tubeName"]));
                        $result = $task->run($data, $jobId);
                        $output->writeln("任务返回:" . $result);
                        if ($result === true) {
                            $beansTalk->delete($job);
                            Redis::getInstance()->redis->setex($cache_key, 300, 1);
                        }
                    } else {
                        $beansTalk->delete($job);
                    }
                }
                usleep(300);
            } catch (ClientException $clientException) {
                Log::error($clientException->getMessage());
            } catch (ServerException $serverException) {
                Log::error($serverException->getMessage());
            } catch (ConnectionException $connectionException) {
                Log::error($connectionException->getMessage());
            } catch (\Exception $exception) {
                $output->writeln("任务异常:" . $exception->getMessage() . $exception->getFile() . $exception->getCode());
                $beansTalk->delete($job);
                Log::error($exception->getMessage());
            }

        }

    }
    //}
    // 父进程等待子进程退出
    /*while (pcntl_waitpid(0, $status) != -1) {
        $status = \pcntl_wexitstatus($status);
    }*/
    //}
}
